package cn.doitedu.operate

import cn.doitedu.util.SparkUtil
import org.apache.spark.sql.expressions.Window

/**
 * @Date 22.4.12
 * @Created by HANGGE
 * @Description
 */
object C08_DSL_Window_Function {
  def main(args: Array[String]): Unit = {
    val session  = SparkUtil.getSession
  // 导入增强
     import  session.implicits._
    import  org.apache.spark.sql.functions._
  //  加载数据
    val df = session.read.option("header", true)
      .option("inferSchema", true)
      .csv("data\\scores\\Scores.csv")
    /**
     * root
     * |-- id: integer (nullable = true)
     * |-- name: string (nullable = true)
     * |-- age: integer (nullable = true)
     * |-- sex: string (nullable = true)
     * |-- city: string (nullable = true)
     * |-- score: double (nullable = true)
     * 统计每个城市下  分数最高的前两个人的信息
     * 1 城市分组
     * 2 对城市内部的分数降序
     * 3 编号
     * 4 获取 编号为 1 2 的信息
     * select
     * *
     * from
     * (
     * select
     * * ,
     * rank() over(partition by city  order by score desc) as  rn
     * from
     * tb_score
     * ) t
     * where  rn <=2
     */

    df.createOrReplaceTempView("tb_score")
 /*   session.sql(
      """
        |select
        |*
        |from
        |(
        |select
        |* ,
        |rank() over(partition by city  order by score desc) as  rn
        |from
        |tb_score
        |) t
        |where  rn <=2
        |""".stripMargin).show()
        +---+----+---+---+----+-----+---+
      | id|name|age|sex|city|score| rn|
      +---+----+---+---+----+-----+---+
      |  9|吕布| 28|  M|深圳| 98.0|  1|
      |  5|曹操| 30|  F|深圳| 90.8|  2|
      |  3|赵云| 20|  F|上海| 88.0|  1|
      |  4|刘备| 26|  M|上海| 83.0|  2|
      |  7|周瑜| 24|  M|北京| 85.0|  1|
      |  2|关羽| 23|  M|北京| 82.0|  2|
      +---+----+---+---+----+-----+---+

        */

    val  wind = Window.partitionBy("city").orderBy('score.desc)

    df.select('id ,'name ,'age,'sex ,'city , 'score ,rank().over(wind) as "rn")
      .where("rn <=2")
      .show()


    df.select('id ,'name ,'age,'sex ,'city , 'score ,rank().over(wind) as "rn")
      .where("rn <=2")
      // 获取指定的字段  子查询
      .select("id" , "name" , "city" , "score")
      .show()
 /*   df.select('id ,'name ,'age,'sex ,'city , 'score ,row_number().over(wind) as "rn")
    df.select('id ,'name ,'age,'sex ,'city , 'score ,collect_set().over(wind) as "rn")
    df.select('id ,'name ,'age,'sex ,'city , 'score ,collect_list().over(wind) as "rn")
    df.select('id ,'name ,'age,'sex ,'city , 'score ,sum().over(wind) as "rn")*/

    df.select('id ,'name ,'age,'sex ,'city , 'score ,rank().over(wind) as "rn")

      .where("rn <=2")
      //  删除原始结果中的一些字段
      .drop("sex")
      .drop("rn")
      .show()

  }

}
